package client

import (
	"sync"
	"sync/atomic"
	"time"

	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/common/logging"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/common/options"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/ichain/events/client/connector"
	eventservice "git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/ichain/events/service"
	esdispatcher "git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/ichain/events/service/dispatcher"
	"git.cloud.inspur.com/ichain/ichain-sdk-go/pkg/providers/ichain"
	"github.com/pkg/errors"
)

var logger = logging.NewLogger("ichainsdk/ichain")

// ConnectionState 客户端连接状态
type ConnectionState int32

const (
	// Disconnected 与远端event service断开
	Disconnected ConnectionState = iota
	// Connecting 正在连接
	Connecting
	// Connected 与event server已连接
	Connected
)

// Client 与inode节点连接，并且可以检测是否需要断开该inode重新连接
type Client struct {
	*eventservice.Service
	params
	sync.RWMutex
	connEvent       chan *connector.ConnectionEvent
	connectionState int32
	stopped         int32
	registerOnce    sync.Once
	afterConnect    handler
	beforeReconnect handler
}

type handler func() error

func New(dispatcher eventservice.Dispatcher, opts ...options.Opt) *Client {
	params := defaultParams()
	options.Apply(params, opts)

	return &Client{
		Service:         eventservice.New(dispatcher, opts...),
		params:          *params,
		connectionState: int32(Disconnected),
	}
}

// SetAfterConnectHandler 注册连接成功后的回调函数
func (c *Client) SetAfterConnectHandler(h handler) {
	c.Lock()
	defer c.Unlock()
	c.afterConnect = h
}

func (c *Client) afterConnectHandler() handler {
	c.RLock()
	defer c.RUnlock()
	return c.afterConnect
}

// SetBeforeReconnectHandler 设置重连前的回调函数
func (c *Client) SetBeforeReconnectHandler(h handler) {
	c.Lock()
	defer c.Unlock()
	c.beforeReconnect = h
}

func (c *Client) beforeReconnectHandler() handler {
	c.RLock()
	defer c.RUnlock()
	return c.beforeReconnect
}

// Connect connects to the inode and registers for events on a particular chain.
func (c *Client) Connect() error {
	if c.maxConnAttempts == 1 {
		return c.connect()
	}
	return c.connectWithRetry(c.maxConnAttempts, c.timeBetweenConnAttempts)
}

// CloseIfIdle 在空闲状态下关闭连接
func (c *Client) CloseIfIdle() bool {
	logger.Debug("Attempting to close event client...")

	regInfoCh := make(chan *esdispatcher.RegistrationInfo)
	err := c.Submit(esdispatcher.NewRegistrationInfoEvent(regInfoCh))
	if err != nil {
		logger.Debugf("Submit failed %s", err)
		return false
	}
	regInfo := <-regInfoCh

	logger.Debugf("Outstanding registrations: %d", regInfo.TotalRegistrations)

	if regInfo.TotalRegistrations > 0 {
		logger.Debugf("Cannot stop client since there are %d outstanding registrations", regInfo.TotalRegistrations)
		return false
	}

	c.Close()

	return true
}

// Close  关闭连接
func (c *Client) Close() {
	c.close(func() {
		c.Stop()
	})
}

func (c *Client) close(stopHandler func()) {
	logger.Debug("Attempting to close event client...")

	if !c.setStoppped() {
		// Already stopped
		logger.Debug("Client already stopped")
		return
	}

	logger.Debug("Stopping client...")

	c.closeConnectEventChan()

	logger.Debug("Sending disconnect request...")

	errch := make(chan error)
	err1 := c.Submit(connector.NewDisconnectEvent(errch))
	if err1 != nil {
		logger.Debugf("Submit failed %s", err1)
		return
	}
	err := <-errch

	if err != nil {
		logger.Warnf("Received error from disconnect request: %s", err)
	} else {
		logger.Debug("Received success from disconnect request")
	}

	logger.Debug("Stopping dispatcher...")

	stopHandler()

	c.mustSetConnectionState(Disconnected)

	logger.Debug("... event client is stopped")
}

// connect 客户端连接
// @return error
func (c *Client) connect() error {
	// 如果客户端已经停止了则返回error
	if c.Stopped() {
		return errors.New("event client is closed")
	}
	// 更新状态断开->正在连接
	if !c.setConnectionState(Disconnected, Connecting) {
		return errors.Errorf("unable to connect event client since client is [%s]. Expecting client to be in state [%s]", c.ConnectionState(), Disconnected)
	}

	logger.Debug("begin summit connection request...")

	errCh := make(chan error, 1)

	// 向event主线程提交连接请求
	if err := c.Submit(connector.NewConnectEvent(errCh)); err != nil {
		return errors.Errorf("Submit failed %s", err)
	}

	// 监听返回
	err := <-errCh
	// 如果执行失败
	if err != nil {
		c.mustSetConnectionState(Disconnected)
		logger.Debugf("... got error in connection response: %s", err)
		return err
	}

	// 单例只执行一次
	c.registerOnce.Do(func() {
		logger.Debug("Submitting connection event registration...")
		_, eventCh, err := c.registerConnectionEvent()
		if err != nil {
			logger.Errorf("Error registering for connection events: %s", err)
			c.Close()
		}
		c.connEvent = eventCh
		go c.monitorConnection()
	})

	// 执行连接回调
	handlerImp := c.afterConnectHandler()
	if handlerImp != nil {
		err = c.execHandler(handlerImp, errCh)
		if err != nil {
			return err
		}
	}

	// 更新连接状态
	c.setConnectionState(Connecting, Connected)

	logger.Debug("Submitting connected event")

	// 提交已连接事件
	err = c.Submit(connector.NewConnectedEvent())
	if err != nil {
		logger.Warnf("Submit failed %s", err)
	}
	return err
}

func (c *Client) execHandler(handlerImp handler, errCh chan error) error {
	// 如果执行出错
	if err := handlerImp(); err != nil {
		logger.Warnf("Error invoking afterConnect handler: %s. Disconnecting...", err)
		// 断开连接
		disconnectErr := c.Submit(connector.NewDisconnectEvent(errCh))
		if disconnectErr != nil {
			logger.Warnf("Submit failed %s", disconnectErr)
		}
		// 监听断开连接是否失败
		select {
		case disconnectErr = <-errCh:
			if disconnectErr != nil {
				logger.Warnf("Received error from disconnect request: %s", disconnectErr)
			} else {
				logger.Debug("Received success from disconnect request")
			}
		case <-time.After(c.respTimeout):
			logger.Warn("Timed out waiting for disconnect response")
		}

		c.setConnectionState(Connecting, Disconnected)

		return errors.WithMessage(err, "error invoking afterConnect handler")
	}
	return nil
}

func (c *Client) connectWithRetry(maxAttempts uint, timeBetweenAttempts time.Duration) error {
	if c.Stopped() {
		return errors.New("event client is closed")
	}
	if timeBetweenAttempts < time.Second {
		timeBetweenAttempts = time.Second
	}

	var attempts uint
	for {
		if c.Stopped() {
			return errors.New("event client is closed")
		}

		attempts++
		logger.Debugf("Attempt #%d to connect...", attempts)
		if err := c.connect(); err != nil {
			logger.Warnf("... connection attempt failed: %s", err)
			if maxAttempts > 0 && attempts >= maxAttempts {
				logger.Warn("maximum connect attempts exceeded")
				return errors.New("maximum connect attempts exceeded")
			}
			time.Sleep(timeBetweenAttempts)
		} else {
			logger.Debug("... connect succeeded.")
			return nil
		}
	}
}

// RegisterBlockEvent 对区块事件的封装
// @param filter
// @return ichain.Registration
// @return <-chan
// @return error
func (c *Client) RegisterBlockEvent(filter ...ichain.BlockFilter) (ichain.Registration, <-chan *ichain.BlockEvent, error) {
	if !c.permitBlockEvents {
		return nil, nil, errors.New("block events are not permitted")
	}
	return c.Service.RegisterBlockEvent(filter...)
}

func (c *Client) registerConnectionEvent() (ichain.Registration, chan *connector.ConnectionEvent, error) {
	if c.Stopped() {
		return nil, nil, errors.New("event client is closed")
	}

	eventch := make(chan *connector.ConnectionEvent, c.eventConsumerBufferSize)
	errch := make(chan error)
	regch := make(chan ichain.Registration)
	err1 := c.Submit(connector.NewRegisterConnectionEvent(eventch, regch, errch))
	if err1 != nil {
		return nil, nil, err1
	}
	select {
	case reg := <-regch:
		return reg, eventch, nil
	case err := <-errch:
		return nil, nil, err
	}
}

func (c *Client) Stopped() bool {
	return atomic.LoadInt32(&c.stopped) == 1
}

func (c *Client) setStoppped() bool {
	return atomic.CompareAndSwapInt32(&c.stopped, 0, 1)
}

func (c *Client) ConnectionState() ConnectionState {
	return ConnectionState(atomic.LoadInt32(&c.connectionState))
}

func (c *Client) setConnectionState(currentState, newState ConnectionState) bool {
	return atomic.CompareAndSwapInt32(&c.connectionState, int32(currentState), int32(newState))
}

func (c *Client) mustSetConnectionState(newState ConnectionState) {
	atomic.StoreInt32(&c.connectionState, int32(newState))
}

func (c *Client) monitorConnection() {
	logger.Debug("Monitoring connection")
	for event := range c.connEvent {
		if c.Stopped() {
			logger.Debugln("Event client has been stopped.")
			break
		}

		c.notifyConnectEventChan(event)

		if event.Connected {
			logger.Debug("Event client has connected")
		} else if c.reconn {
			logger.Warnf("Event client has disconnected. Details: %s", event.Err)
			if c.setConnectionState(Connected, Disconnected) {
				if event.Err.IsFatal() {
					logger.Warnf("Reconnect is not possible due to fatal error. Terminating: %s", event.Err)
					go c.Close()
					break
				}

				logger.Warn("Attempting to reconnect...")
				go c.reconnect()
			} else if c.setConnectionState(Connecting, Disconnected) {
				logger.Warn("Reconnect already in progress. Setting state to disconnected")
			}
		} else {
			logger.Debugf("Event client has disconnected. Terminating: %s", event.Err)
			go c.Close()
			break
		}
	}
	logger.Debug("Exiting connection monitor")
}

func (c *Client) reconnect() {
	logger.Debugf("Waiting %s before attempting to reconnect event client...", c.reconnInitialDelay)
	time.Sleep(c.reconnInitialDelay)

	logger.Debug("Attempting to reconnect event client...")

	handlerImp := c.beforeReconnectHandler()
	if handlerImp != nil {
		if err := handlerImp(); err != nil {
			logger.Errorf("Error invoking beforeReconnect handler: %s", err)
			return
		}
	}

	if err := c.connectWithRetry(c.maxReconnAttempts, c.timeBetweenConnAttempts); err != nil {
		logger.Warnf("Could not reconnect event client: %s", err)
		if !c.Stopped() {
			c.Close()
		}
	} else {
		logger.Infof("Event client has reconnected")
	}
}

func (c *Client) closeConnectEventChan() {
	c.Lock()
	defer c.Unlock()
	if c.connEventCh != nil {
		close(c.connEventCh)
	}
}

func (c *Client) notifyConnectEventChan(event *connector.ConnectionEvent) {
	c.RLock()
	defer c.RUnlock()
	if c.connEventCh != nil {
		logger.Debugln("Sending connection event to subscriber.")
		c.connEventCh <- event
	}
}

func (s ConnectionState) String() string {
	switch s {
	case Disconnected:
		return "Disconnected"
	case Connected:
		return "Connected"
	case Connecting:
		return "Connecting"
	default:
		return "undefined"
	}
}
